import copy
import json
import os
import queue
import sys
import time
import threading
from loguru import logger
from datetime import datetime, timedelta
from kafka import KafkaConsumer, KafkaProducer, TopicPartition
from kafka.errors import NoBrokersAvailable, KafkaError


dir = os.path.dirname(os.path.abspath(__file__))
log_dir = os.path.join(dir, "Log", "logs")
if not os.path.exists(log_dir):
    os.makedirs(log_dir)
# 自定义日志格式
logger.remove()  # 移除默认的日志处理器
logger.add(sink=sys.stdout, format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {message}", level="INFO")
logger.add(
    sink=os.path.join(log_dir, "clean_{time:YYYY-MM-DD}.log"),  # 日志文件名带日期
    format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {function}:{line} - {message}",
    rotation="00:00",           # 每天午夜轮转（与大小轮转二选一）
    retention="360 days",       # 保留最近360天的日志
    encoding="utf-8",           # 避免中文乱码
    level="INFO",               # 文件日志级别
)
kako_flag = False
# 中国地区简称列表（省级行政区）
REGION_PREFIX = {
    '京', '津', '冀', '晋', '蒙', '辽', '吉', '黑',
    '沪', '苏', '浙', '皖', '闽', '赣', '鲁', '豫',
    '鄂', '湘', '粤', '桂', '琼', '渝', '川', '贵',
    '云', '藏', '陕', '甘', '青', '宁', '新'
}


class SichuanClean:
    def __init__(self, device_info, config_info):
        # 设备信息
        self.device_info = device_info
        # 配置信息
        self.config_data = config_info["ConfigData"]
        # 计算精度的时间区间，默认1小时
        self.duration_time = float(self.config_data["clean_duration_time"])
        # kafka生产者
        self.kafka_producer = None
        # 接收上下行重识别数据保存的队列
        self.q = queue.Queue(maxsize=1000)
        # 接收设备心跳信息保存队列
        self.q_ = queue.Queue(maxsize=1000)
        # kafka 连接状态,1-异常，0-正常
        self.kafka_status = 1
        # 车流数据根据点位缓存
        self.q_dict = {}
        # 设备状态缓存
        self.clock_status = {}
        # 设备处理结果缓存
        self.device_result_dict = {}
        # 摄像机IP-设备ID对应字典
        self.cameraIP_device_dict = {}
        # 上下行设备路由列表
        self.up_device_list = device_info["route_info"]["up"]
        self.down_device_list = device_info["route_info"]["down"]
        # 上下行全域延迟时间-ms
        self.up_delay_ms = None
        self.down_delay_ms = None
        # 卡口数据处理
        self.q_kako = queue.Queue(maxsize=1000)
        self.kako_dict = {}
        self.kako_delay_ms_dict = {}
        self.kako_device_list = device_info["route_info"]["kakou"]["up"] + device_info["route_info"]["kakou"]["down"]

    def deal_data(self, vehicle_info):
        """
        处理上下行数据
        """
        try:
            orgCode = vehicle_info.get('orgCode', "")
            globalTime = vehicle_info.get('globalTime', 0)
            if "targetList" in vehicle_info:
                for target in vehicle_info["targetList"]:
                    receiveTime = vehicle_info.get('receiveTime', 0)
                    sendTime = vehicle_info.get('sendTime', 0)
                    id = target.get("id", 0)
                    carType = target.get("carType", 0)
                    carColor = target.get("carColor", 0)
                    station = str(target.get("station", 0))
                    new_id = f"{id}_{carType}_{carColor}"
                    if station == "0" or id == 0 or receiveTime == 0 or sendTime == 0:
                        continue
                    vehicle_data = {
                        "device": station,
                        "time": globalTime,
                        "id": id,
                        "type": carType,
                        "color": carColor,
                        "Id": new_id
                    }
                    self.clock_status[station]["receive_time"] = receiveTime
                    self.clock_status[station]["send_time"] = sendTime
                    self.clock_status[station]["global_time"] = globalTime
                    if station in self.up_device_list and self.up_delay_ms is None:
                        self.up_delay_ms = globalTime - receiveTime
                        logger.info(f"上行时钟延迟: {self.up_delay_ms} ms")
                    elif station in self.down_device_list and self.down_delay_ms is None:
                        self.down_delay_ms = globalTime - receiveTime
                        logger.info(f"下行时钟延迟: {self.down_delay_ms} ms")
                    else:
                        pass
                    self.q.put(vehicle_data, block=False)
        except queue.Full:
            self.q.get(block=False)  # 手动丢弃队头数据（1）
            logger.warning("队列已满，丢弃队头数据")

    def deal_kako_data(self, vehicle_info):
        """
        处理卡口数据
        """
        formats = [
            '%Y-%m-%dT%H:%M:%S.%f',  # ISO 8601（带毫秒）
            '%Y-%m-%d %H:%M:%S.%f',  # 空格分隔（带毫秒）
            '%Y-%m-%dT%H:%M:%S',  # ISO 8601（无毫秒）
            '%Y-%m-%d %H:%M:%S',  # 空格分隔（无毫秒）
            '%Y/%m/%d %H:%M:%S',  # 斜杠分隔
        ]
        try:
            timestamp = vehicle_info.get('timestamp', "")
            if timestamp == "":
                return
            dt = None
            for fmt in formats:
                try:
                    dt = datetime.strptime(timestamp, fmt)
                    break
                except ValueError:
                    continue
            if dt is None:
                return
            ts = int(dt.timestamp() * 1000)
            sn = vehicle_info.get('sn', "")
            # ip = vehicle_info.get('ip', "")
            plateNumber = vehicle_info.get('plateNumber', "").strip()
            vehicleType = vehicle_info.get('vehicleType', 0)
            sendTime = vehicle_info.get('sendTime', 0)
            receiveTime = vehicle_info.get('receiveTime', 0)
            if plateNumber:
                # if plateNumber is None or plateNumber == "车牌" or plateNumber == "" or "默" in plateNumber:
                #     pass
                vehicle_data = {
                    "device": sn,
                    "global_time": ts,
                    "receive_time": receiveTime,
                    "send_time": sendTime,
                    "id": plateNumber,
                    "type": vehicleType
                }
                self.clock_status[sn]["receive_time"] = receiveTime
                self.clock_status[sn]["send_time"] = sendTime
                self.clock_status[sn]["global_time"] = ts
                if self.kako_delay_ms_dict[sn] is None:
                    self.kako_delay_ms_dict[sn] = ts - receiveTime
                    logger.info(f"卡口{sn}时钟延迟: {self.kako_delay_ms_dict[sn]} ms")
                self.q_kako.put(vehicle_data, block=False)

        except queue.Full:
            self.q_kako.get(block=False)  # 手动丢弃队头数据（1）
            logger.warning("队列已满，丢弃队头数据")

    def deal_heartbeat(self, heartbeat_info):
        """
        处理设备心跳数据
        """
        try:
            timeStamp = heartbeat_info["timeStamp"]
            timestamp = datetime.strptime(timeStamp, "%Y-%m-%d %H:%M:%S:%f")
            cameraList = heartbeat_info["cameraList"]
            status_data = {"timestamp": timestamp, "camera_list": cameraList}
            self.q_.put(status_data, block=False)
        except queue.Full:
            self.q_.get(block=False)  # 手动丢弃队头数据（1）
            logger.warning("队列已满，丢弃队头数据")

    def subscribe_kafka(self, topic, callbacks, partition=None):
        """
        kafak重连机制，实时监听数据，并通过callbacks进行数据处理
        """
        kafka_consumer = None
        while True:
            time.sleep(0.002)
            # kafka消费者连接，失败会自动重连
            if kafka_consumer is None:
                try:
                    if partition is None:
                        kafka_consumer = KafkaConsumer(
                            topic,
                            bootstrap_servers=self.config_data["kafka_host"],
                            auto_offset_reset='latest',
                            enable_auto_commit=False
                        )
                        self.kafka_status = 0
                        logger.info(f"原始数据监听启动, topic:{topic}")
                    else:
                        kafka_consumer = KafkaConsumer(
                            topic,
                            bootstrap_servers=self.config_data["kafka_host"],
                            auto_offset_reset='latest',
                            enable_auto_commit=False
                        )
                        self.kafka_status = 0
                        # 绑定到指定分区
                        tp = TopicPartition(topic, partition)
                        kafka_consumer.assign([tp])
                        logger.info(f"原始数据监听启动, topic:{topic}, partition:{partition}")
                except NoBrokersAvailable as e:
                    logger.error(f"kafka连接失败, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    continue
            while True:
                time.sleep(0.0002)
                if self.kafka_status == 1:
                    logger.error("kafka连接断开")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                try:
                    records = kafka_consumer.poll(timeout_ms=1000)
                    if records:
                        for tp, messages in records.items():
                            for message in messages:
                                if message:
                                    # 数据处理，卡口数据/重识别数据
                                    message_value = message.value.decode('utf-8')
                                    vehicle_info = json.loads(message_value)
                                    callbacks(vehicle_info)
                except KafkaError as e:
                    # kafka异常处理
                    logger.error(f"kafka连接断开, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                except Exception as e:
                    logger.opt(exception=e).error(f"解析数据失败, topic:{topic}")
                    pass

    def subscribe_up_data_kafka(self):
        """
        监听kafka数据，支持断线重连
        """
        topic = self.config_data["kafka_up_cars_topic"]
        kafka_consumer = None
        while True:
            time.sleep(0.002)
            if kafka_consumer is None:
                try:
                    kafka_consumer = KafkaConsumer(
                        topic,
                        bootstrap_servers=self.config_data["kafka_host"],
                        auto_offset_reset='latest',
                        enable_auto_commit=False
                    )
                    self.kafka_status = 0
                    logger.info(f"上行门架数据监听启动, topic:{topic}")
                except NoBrokersAvailable as e:
                    logger.error(f"kafka连接失败, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    continue

            while True:
                time.sleep(0.0002)
                if self.kafka_status == 1:
                    logger.error("kafka连接断开")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                try:
                    records = kafka_consumer.poll(timeout_ms=1000)
                    if records:
                        for tp, messages in records.items():
                            for message in messages:
                                if message:
                                    message_value = message.value.decode('utf-8')
                                    vehicle_info = json.loads(message_value)
                                    orgCode = vehicle_info.get('orgCode', "")
                                    # receiveTime = vehicle_info.get('receiveTime', 0)
                                    # sendTime = vehicle_info.get('sendTime', 0)
                                    globalTime = vehicle_info.get('globalTime', 0)
                                    if "targetList" in vehicle_info:
                                        for target in vehicle_info["targetList"]:
                                            receiveTime = vehicle_info.get('receiveTime', 0)
                                            sendTime = vehicle_info.get('sendTime', 0)
                                            id = target.get("id", 0)
                                            carType = target.get("carType", 0)
                                            carColor = target.get("carColor", 0)
                                            station = str(target.get("station", 0))
                                            # station = target.get("sn", 0)
                                            new_id = f"{id}_{carType}_{carColor}"
                                            if station == "0" or id == 0 or receiveTime == 0 or sendTime == 0:
                                                continue
                                            vehicle_data = {
                                                "device": station,
                                                "time": globalTime,
                                                "id": id,
                                                "type": carType,
                                                "color": carColor,
                                                "Id": new_id
                                            }
                                            self.clock_status[station]["receive_time"] = receiveTime
                                            self.clock_status[station]["send_time"] = sendTime
                                            self.clock_status[station]["global_time"] = globalTime
                                            if self.up_delay_ms is None:
                                                self.up_delay_ms = globalTime - receiveTime
                                            self.q.put(vehicle_data, block=False)
                except queue.Full:
                    self.q.get(block=False)  # 手动丢弃队头数据（1）
                    logger.warning("队列已满，丢弃队头数据")
                except KafkaError as e:
                    logger.error(f"kafka连接断开, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                except Exception as e:
                    logger.opt(exception=e).error(f"解析数据失败, topic:{topic}")
                    pass

    def subscribe_down_data_kafka(self):
        """
        监听kafka数据，支持断线重连
        """
        topic = self.config_data["kafka_down_cars_topic"]
        kafka_consumer = None
        while True:
            time.sleep(0.002)
            if kafka_consumer is None:
                try:
                    kafka_consumer = KafkaConsumer(
                        topic,
                        bootstrap_servers=self.config_data["kafka_host"],
                        auto_offset_reset='latest',
                        enable_auto_commit=False
                    )
                    self.kafka_status = 0
                    logger.info(f"下行门架数据监听启动, topic:{topic}")
                except NoBrokersAvailable as e:
                    logger.error(f"kafka连接失败, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    continue

            while True:
                time.sleep(0.0002)
                if self.kafka_status == 1:
                    logger.error("kafka连接断开")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                try:
                    records = kafka_consumer.poll(timeout_ms=1000)
                    if records:
                        for tp, messages in records.items():
                            for message in messages:
                                if message:
                                    message_value = message.value.decode('utf-8')
                                    vehicle_info = json.loads(message_value)
                                    orgCode = vehicle_info.get('orgCode', "")
                                    receiveTime = vehicle_info.get('receiveTime', 0)
                                    sendTime = vehicle_info.get('sendTime', 0)
                                    globalTime = vehicle_info.get('globalTime', 0)
                                    if "targetList" in vehicle_info:
                                        for target in vehicle_info["targetList"]:
                                            id = target.get("id", 0)
                                            carType = target.get("carType", 0)
                                            carColor = target.get("carColor", 0)
                                            station = str(target.get("station", 0))
                                            new_id = f"{id}_{carType}_{carColor}"
                                            if station == "0" or id == 0 or receiveTime == 0 or sendTime == 0:
                                                continue
                                            vehicle_data = {
                                                "device": station,
                                                "time": receiveTime,
                                                "id": id,
                                                "type": carType,
                                                "color": carColor,
                                                "Id": new_id
                                            }
                                            if station == "0":
                                                logger.warning(f"staion异常, data:{message_value}")
                                                continue
                                            try:
                                                self.clock_status[station]["receive_time"] = receiveTime
                                                self.clock_status[station]["send_time"] = sendTime
                                                self.clock_status[station]["global_time"] = globalTime
                                                if self.down_delay_ms is None:
                                                    self.down_delay_ms = globalTime - receiveTime
                                                self.q.put(vehicle_data, block=False)
                                            except Exception as e:
                                                logger.opt(exception=e).error(f"解析数据失败, topic:{topic}")
                                                pass
                except queue.Full:
                    self.q.get(block=False)  # 手动丢弃队头数据（1）
                    logger.warning("队列已满，丢弃队头数据")
                except KafkaError as e:
                    logger.error(f"kafka连接断开, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                except Exception as e:
                    logger.opt(exception=e).error(f"解析数据失败, topic:{topic}")
                    pass

    def subscribe_status_kafka(self):
        """
        监听kafka数据，支持断线重连
        """
        topic = self.config_data["kafka_status_topic"]
        kafka_consumer = None
        while True:
            time.sleep(0.002)
            if kafka_consumer is None:
                try:
                    kafka_consumer = KafkaConsumer(
                        topic,
                        bootstrap_servers=self.config_data["kafka_host"],
                        auto_offset_reset='latest',
                        enable_auto_commit=False
                    )
                    self.kafka_status = 0
                    logger.info(f"下行门架数据监听启动, topic:{topic}")
                except NoBrokersAvailable as e:
                    logger.error(f"kafka连接失败, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    continue

            while True:
                time.sleep(0.0002)
                if self.kafka_status == 1:
                    logger.error("kafka连接断开")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                try:
                    records = kafka_consumer.poll(timeout_ms=1000)
                    if records:
                        for tp, messages in records.items():
                            for message in messages:
                                if message:
                                    message_value = message.value.decode('utf-8')
                                    status_info = json.loads(message_value)
                                    timeStamp = status_info["timeStamp"]
                                    timestamp = datetime.strptime(timeStamp, "%Y-%m-%d %H:%M:%S:%f")
                                    cameraList = status_info["cameraList"]
                                    status_data = {"timestamp": timestamp,
                                                   "camera_list": cameraList}
                                    self.q_.put(status_data, block=False)
                except queue.Full:
                    self.q_.get(block=False)  # 手动丢弃队头数据（1）
                    logger.warning("队列已满，丢弃队头数据")
                except KafkaError as e:
                    logger.error(f"kafka连接断开, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                except Exception as e:
                    logger.opt(exception=e).error(f"解析数据失败, topic:{topic}")
                    pass

    def init(self):
        all_device_list = self.up_device_list + self.down_device_list
        for i in all_device_list:
            self.q_dict[i] = []
            self.clock_status[i] = {"receive_time": 0, "send_time": 0, "global_time": 0}
            self.device_result_dict[i] = {"total": 0, "positive": 0, "missed": 0, "incorrect": 0, "accuracy": 0,
                                          "heartbeat": 1, "clock": 1, "delay": 1}
        logger.info(f"设备列表:{list(self.clock_status.keys())}")
        if kako_flag:
            for j in self.kako_device_list:
                self.kako_dict[j] = []
                self.kako_delay_ms_dict[j] = None
                self.clock_status[j] = {"receive_time": 0, "send_time": 0, "global_time": 0}
                self.device_result_dict[j] = {"total": 0, "positive": 0, "missed": 0, "incorrect": 0, "accuracy": 0,
                                              "heartbeat": 1, "clock": 1, "delay": 1}
        for key, devices in self.device_info["device_info"].items():
            for id, deviceInfo in devices.items():
                self.cameraIP_device_dict[deviceInfo["cameraIP"]] = id
        logger.info(f"设备列表:{self.cameraIP_device_dict}")
        # TODO 启动线程实时监听kafka数据，根据数据情况适当调整，最终数据完整时将下面三个线程全打开
        # threading.Thread(target=self.subscribe_up_data_kafka).start()
        # threading.Thread(target=self.subscribe_down_data_kafka).start()
        # threading.Thread(target=self.subscribe_status_kafka).start()
        # TODO 优化数据监听结构，如果数据量大则需要根据kafka分区分别监听，现在是监听所有分区
        up_topic = self.config_data["kafka_up_cars_topic"]
        down_topic = self.config_data["kafka_down_cars_topic"]
        status_topic = self.config_data["kafka_status_topic"]
        threading.Thread(target=self.subscribe_kafka, args=(up_topic, self.deal_data,)).start()
        threading.Thread(target=self.subscribe_kafka, args=(down_topic, self.deal_data,)).start()
        threading.Thread(target=self.subscribe_kafka, args=(status_topic, self.deal_heartbeat,)).start()
        if kako_flag:
            kako_topic = self.config_data["kafka_kakou_cars_topic"]
            threading.Thread(target=self.subscribe_kafka, args=(kako_topic, self.deal_kako_data,)).start()

    def hour_data(self, vehicle_list):
        k = 0
        if vehicle_list:
            current_time = vehicle_list[-1]["time"]
            for i in range(len(vehicle_list)):
                vehicle_time = vehicle_list[i]["time"]
                if vehicle_time < current_time - 60 * 60 * 1000 * self.duration_time:
                    continue
                else:
                    k = i
                    break
        return k

    def run(self):
        # 循环监听心跳，初始化时间
        last_time = time.time()
        while True:
            time.sleep(0.0002)
            try:
                # kafka 服务器
                if self.kafka_producer is None:
                    logger.info(f"add_message {self.kafka_producer}")
                    try:
                        self.kafka_producer = KafkaProducer(
                            bootstrap_servers=self.config_data["kafka_host"],
                            request_timeout_ms=10000,
                            value_serializer=lambda x: json.dumps(x).encode('utf-8')
                        )
                        self.kafka_status = 0
                        logger.info(f"kafka_producer连接成功")
                    except NoBrokersAvailable as e:
                        self.kafka_status = 1
                        logger.error("kafka连接失败")
                        self.kafka_producer = None
                        time.sleep(10)
                        continue
                # 心跳状态上报
                if time.time() - last_time >= 1:
                    m = {
                        "timestamp": str(int(time.time())),
                        "status": 0,
                        "type": 1
                    }
                    try:
                        future = self.kafka_producer.send(self.config_data["kafka_status"], value=m)
                        record_metadata = future.get(timeout=10)  # 正常情况下很快返回
                    except KafkaError as e:
                        self.kafka_status = 1
                        logger.error("kafka连接失败，导致算法心跳上报失败")
                        self.kafka_producer = None
                        time.sleep(10)
                    except Exception as e:
                        logger.error("算法心跳上报失败")
                    # 更新时间
                    last_time = time.time()

                # 从队列中获取最新车流数据
                while not self.q.empty():
                    vehicle_info = self.q.get(block=False)
                    self.q_dict[vehicle_info["device"]].append(vehicle_info)
                # 从队列中获取最新卡口数据
                while not self.q_kako.empty():
                    kako_info = self.q_kako.get(block=False)
                    self.kako_dict[kako_info["device"]].append(kako_info["id"])
                # 从队列中获取最新设备状态数据
                while not self.q_.empty():
                    status_info = self.q_.get(block=False)
                    for camera in status_info["camera_list"]:
                        if camera["ip"] not in self.cameraIP_device_dict:
                            continue
                        camera_id = self.cameraIP_device_dict[camera["ip"]]
                        camera_status = str(camera["status"])
                        camera_status_list = camera_status.split(",")
                        if "2" in camera_status_list:
                            self.device_result_dict[camera_id]["heartbeat"] = 1
                        elif camera_status == "1":
                            self.device_result_dict[camera_id]["heartbeat"] = 0
                        else:
                            self.device_result_dict[camera_id]["heartbeat"] = camera_status
                # 设备状态数据处理，并发送kafka
                if int(time.time()) % 10 == 0:
                    time.sleep(1)
                    # 只保留各基站位置最近一个小时数据进行数据诊断
                    for device_id, vehicle_list in self.q_dict.items():
                        del_index = self.hour_data(vehicle_list)
                        self.q_dict[device_id] = vehicle_list[del_index:]
                        # 获取基站device_id在最近一段时间内的过车总数
                        self.device_result_dict[device_id]["total"] = len(self.q_dict[device_id])

                    # 上行设备精度计算
                    for i in range(len(self.up_device_list) - 1):
                        # if i == 0:
                        #     self.device_result_dict[self.up_device_list[0]]["positive"] = self.device_result_dict[self.up_device_list[0]]["total"]
                        #     continue
                        positive_id_dict = []
                        up_device_id = self.up_device_list[i]
                        down_device_id = self.up_device_list[i + 1]
                        up_device_data_dict = {vehicle_info["Id"]: vehicle_info for vehicle_info in self.q_dict[up_device_id]}
                        down_device_data_dict = {vehicle_info["Id"]: vehicle_info for vehicle_info in self.q_dict[down_device_id]}
                        for car_id, down_vehicle_info in down_device_data_dict.items():
                            if car_id in up_device_data_dict.keys():
                                positive_id_dict.append(car_id)
                                self.device_result_dict[down_device_id]["positive"] += 1
                        num = self.device_result_dict[down_device_id]["total"] - len(positive_id_dict)
                        self.device_result_dict[down_device_id]["incorrect"] += num
                        # 计算上下游的漏检
                        num_diff = self.device_result_dict[up_device_id]["total"] - self.device_result_dict[down_device_id]["total"]
                        if num_diff > 0:
                            self.device_result_dict[down_device_id]["missed"] += num_diff

                    # 下行设备精度计算
                    for i in range(len(self.down_device_list) - 1):
                        if i == 0:
                            self.device_result_dict[self.down_device_list[0]]["positive"] = self.device_result_dict[self.down_device_list[0]]["total"]
                            continue
                        positive_id_dict = []
                        up_device_id = self.down_device_list[i]
                        down_device_id = self.down_device_list[i + 1]
                        up_device_data_dict = {vehicle_info["Id"]: vehicle_info for vehicle_info in self.q_dict[up_device_id]}
                        down_device_data_dict = {vehicle_info["Id"]: vehicle_info for vehicle_info in self.q_dict[down_device_id]}
                        for car_id, up_vehicle_info in up_device_data_dict.items():
                            if car_id in down_device_data_dict.keys():
                                positive_id_dict.append(car_id)
                                self.device_result_dict[down_device_id]["positive"] += 1
                        num = self.device_result_dict[down_device_id]["total"] - len(positive_id_dict)
                        self.device_result_dict[down_device_id]["incorrect"] += num
                        # 计算上下游的漏检
                        num_diff = self.device_result_dict[up_device_id]["total"] - self.device_result_dict[down_device_id]["total"]
                        if num_diff > 0:
                            self.device_result_dict[down_device_id]["missed"] += num_diff

                    if kako_flag:
                        # 卡口数据计算
                        for kako_id, kako_data in self.kako_dict.items():
                            # 计算有效数据数量（排除"车牌"、空字符串、包含"默"和首字母非地区简称的数据）
                            valid_count = sum(
                                1 for item in kako_data
                                if item not in ("车牌", "") and
                                "默" not in item and
                                len(item) > 0 and
                                item[0] in REGION_PREFIX
                            )
                            # 计算准确率（避免除零错误）
                            all_num = len(kako_data)
                            # kako_accuracy[kako_id] = valid_count / all_num * 100 if all_num > 0 else 100
                            self.device_result_dict[kako_id]["total"] = all_num
                            self.device_result_dict[kako_id]["positive"] = valid_count
                            # 限制列表长度不超过1000（使用切片更高效）
                            if len(kako_data) > 1000:
                                self.kako_dict[kako_id] = kako_data[-1000:]

                    # 计算基站状态并发送到kafka
                    result = {"timestamp": int(time.time()), "devices": []}
                    for device_id, data in self.device_result_dict.items():
                        if data["total"] == 0:
                            accuracy = 100
                        else:
                            accuracy = data["positive"] / data["total"] * 100
                        self.device_result_dict[device_id]["accuracy"] = round(accuracy, 2)
                        if device_id == self.up_device_list[0] or device_id == self.down_device_list[0]:
                            self.device_result_dict[device_id]["positive"] = self.device_result_dict[device_id]["total"]
                            self.device_result_dict[device_id]["accuracy"] = 100
                        # logger.info(f"设备{device_id}的精度为{accuracy}")

                        clock_status = self.clock_status[device_id]
                        if device_id in self.up_device_list:
                            delay_ms = self.up_delay_ms if self.up_delay_ms is not None else 0
                        elif device_id in self.down_device_list:
                            delay_ms = self.down_delay_ms if self.down_delay_ms is not None else 0
                        else:
                            delay_ms = self.kako_delay_ms_dict[device_id] if device_id in self.kako_delay_ms_dict else 0
                            if delay_ms is None:
                                delay_ms = 0
                        delay_time = abs(clock_status["global_time"] - clock_status["send_time"] - delay_ms) if clock_status["send_time"] != 0 else 0
                        clock_time = abs(clock_status["global_time"] - clock_status["receive_time"] - delay_ms) if clock_status["send_time"] != 0 else 0
                        # mm = clock_status["global_time"] - clock_status["send_time"]
                        # logger.info(f"{device_id}设备的延时{clock_status}, {mm}, {delay_ms}")
                        # TODO 延迟和时钟偏差大于5s则认为是异常
                        self.device_result_dict[device_id]["delay"] = 1 if delay_time > self.config_data["delay_time"] else 0
                        self.device_result_dict[device_id]["clock"] = 1 if clock_time > self.config_data["clock_time"] else 0
                        # 如果心跳丢失则认为设备参数全部异常
                        # if self.device_result_dict[device_id]["heartbeat"] == 1:
                        #     self.device_result_dict[device_id]["delay"] = 1
                        #     self.device_result_dict[device_id]["clock"] = 1
                        #     self.device_result_dict[device_id]["accuracy"] = 0
                        result_device = {"deviceId": device_id,
                                         "heartbeat": self.device_result_dict[device_id]["heartbeat"],
                                         "clock": self.device_result_dict[device_id]["delay"],
                                         "delay": self.device_result_dict[device_id]["clock"],
                                         "precision": self.device_result_dict[device_id]["accuracy"],
                                         "total": self.device_result_dict[device_id]["total"],
                                         "positive": self.device_result_dict[device_id]["positive"],
                                         "missed": self.device_result_dict[device_id]["missed"],
                                         "incorrect": self.device_result_dict[device_id]["incorrect"]}
                        result["devices"].append(result_device)
                    self.kafka_producer.send(self.config_data["kafka_device_status"], value=result)
                    logger.info(f"设备状态发送成功, {result}")
                    for device_id in self.device_result_dict:
                        # self.clock_status[device_id] = {"receive_time": 0, "send_time": 0, "global_time": 0}
                        # 这里更改的原因是心跳数据发送频率不是整10秒，而是10秒多一点，所以不能用整10秒的间隔，因此这里更改为：心跳数据使用上一次的数据，有新的数据后再更新
                        # self.device_result_dict[device_id] = {"total": 0, "positive": 0, "missed": 0, "incorrect": 0,
                        #                                       "accuracy": 0, "heartbeat": 1, "clock": 1, "delay": 1}
                        self.device_result_dict[device_id]["total"] = 0
                        self.device_result_dict[device_id]["positive"] = 0
                        self.device_result_dict[device_id]["missed"] = 0
                        self.device_result_dict[device_id]["incorrect"] = 0
                        self.device_result_dict[device_id]["accuracy"] = 0
                        self.device_result_dict[device_id]["clock"] = 0
                        self.device_result_dict[device_id]["delay"] = 0

            except Exception as e:
                logger.opt(exception=e).error("数据处理异常")


if __name__ == '__main__':
    # 获取当前文件所在目录
    current_dir = os.path.dirname(__file__)
    print(current_dir)
    # 获取相机设备文件信息
    device_file_path = os.path.join(current_dir, 'Config', 'device_info.json')
    with open(device_file_path, 'r', encoding='utf-8') as f:
        device_info = json.load(f)
    # print(device_info)
    # 获取配置文件信息
    config_file_path = os.path.join(current_dir, 'Config', 'config.json')
    with open(config_file_path, 'r', encoding='utf-8') as ff:
        config_info = json.load(ff)
    # print(config_info)
    # 创建对象
    sichuan_clean = SichuanClean(device_info, config_info)
    sichuan_clean.init()
    sichuan_clean.run()
